package com.bridgeintelligent.tag.bulkload.service.core.append;

import com.bridgeintelligent.tag.bulkload.api.BulkLoadConfig;
import com.bridgeintelligent.tag.bulkload.service.core.refresh.BulkRefreshApi;
import com.bridgeintelligent.tag.bulkload.service.core.refresh.BulkRefreshApiUtils;
import com.bridgeintelligent.tag.bulkload.service.core.refresh.BulkRefreshConfig;
import com.bridgeintelligent.tag.mapper.EsIndexNameMapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@AllArgsConstructor
@Slf4j
@Component
//@RocketMQMessageListener(topic = "DIM_APPEND_TOP", consumerGroup = "DIM_APPEND_GROUP", consumeThreadMax=1)
public class DimAppendDelegate implements RocketMQListener<String> {

    private ObjectMapper objectMapper;
    private EsIndexNameMapper esIndexNameMapper;
    private BulkRefreshConfig bulkRefreshConfig;
    private BulkRefreshApiUtils bulkRefreshApiUtils;
    private DimAppendHandler dimAppendHandler;


    @SneakyThrows
    @Override
    public void onMessage(String message) {
        long time = System.currentTimeMillis();
        try {
            log.info("历史数据追加新标签topic收到消息=======>{}", message);
            DimAppendMqMsg dimAppendMqMsg = objectMapper.readValue(message, DimAppendMqMsg.class);
            //设置索引类型 初始化数据为1
            String indexType = esIndexNameMapper.findIndexType("1");
            dimAppendMqMsg.setIndexType(indexType);
            //处理消息
            log.debug("\n------------->>>>开始处理消息");
            dimAppendHandler.handler(dimAppendMqMsg);
            log.debug("\n------------->>>>处理完消息");
        } catch (Exception e) {
            log.error("历史数据追加新标签消费消息时出现异常=======>", e);
        } finally {
            refreshEngineCache();
            log.debug("消费历史数据追加新标签数据总耗时=====>{}ms", (System.currentTimeMillis() - time));
        }
    }

    /**
     * 刷新引擎缓存
     */
    private void refreshEngineCache() {
        String url = bulkRefreshConfig.getRequestUrl();
        BulkRefreshApi bulkRefreshApi = bulkRefreshApiUtils.build(BulkRefreshApi.class, url + BulkLoadConfig.REFRESH_PATH);
        log.info("刷新引擎地址："+url+BulkLoadConfig.REFRESH_PATH);
        bulkRefreshApi.refreshItem();
    }

}
